package com.chukun.flink.stream.action.fraud.rules.functions;

import com.chukun.flink.stream.action.fraud.rules.major.Descriptors;
import com.chukun.flink.stream.action.fraud.rules.utils.KeysExtractor;
import com.chukun.flink.stream.action.fraud.rules.model.Keyed;
import com.chukun.flink.stream.action.fraud.rules.model.Rule;
import com.chukun.flink.stream.action.fraud.rules.model.Transaction;
import com.chukun.flink.stream.action.fraud.rules.utils.ProcessingUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;

/**
 * @author chukun
 * @version 1.0.0
 * @description 处理key的函数
 * @createTime 2022年05月22日 16:14:00
 */
public class DynamicKeyFunction extends BroadcastProcessFunction<Transaction, Rule, Keyed<Transaction, String, Integer>> {

    private static final Logger logger = LoggerFactory.getLogger(DynamicKeyFunction.class);

    private RuleCounterGauge ruleCounterGauge;

    @Override
    public void open(Configuration parameters) throws Exception {
        ruleCounterGauge = new RuleCounterGauge();
        getRuntimeContext().getMetricGroup().gauge("numberOfActiveRules", ruleCounterGauge);
    }

    @Override
    public void processElement(Transaction transaction, BroadcastProcessFunction<Transaction, Rule, Keyed<Transaction, String, Integer>>.ReadOnlyContext readOnlyContext, Collector<Keyed<Transaction, String, Integer>> collector) throws Exception {
        ReadOnlyBroadcastState<Integer, Rule> rulesState = readOnlyContext.getBroadcastState(Descriptors.rulesDescriptor);
        forkEventForEachGroupingKey(transaction, rulesState, collector);
    }

    /**
     * 生成key
     * @param event
     * @param rulesState
     * @param out
     * @throws Exception
     */
    private void forkEventForEachGroupingKey(
            Transaction event,
            ReadOnlyBroadcastState<Integer, Rule> rulesState,
            Collector<Keyed<Transaction, String, Integer>> out)
            throws Exception {
        int ruleCounter = 0;
        for (Map.Entry<Integer, Rule> entry : rulesState.immutableEntries()) {
            final Rule rule = entry.getValue();
            out.collect(
                    new Keyed<>(
                            event, KeysExtractor.getKey(rule.getGroupingKeyNames(), event), rule.getRuleId()));
            ruleCounter++;
        }
        ruleCounterGauge.setValue(ruleCounter);
    }

    @Override
    public void processBroadcastElement(Rule rule, BroadcastProcessFunction<Transaction, Rule, Keyed<Transaction, String, Integer>>.Context context, Collector<Keyed<Transaction, String, Integer>> collector) throws Exception {
        logger.trace("Processing {}", rule);
        BroadcastState<Integer, Rule> broadcastState =
                context.getBroadcastState(Descriptors.rulesDescriptor);
        ProcessingUtils.handleRuleBroadcast(rule, broadcastState);
        if (rule.getRuleState() == Rule.RuleState.CONTROL) {
            handleControlCommand(rule.getControlType(), broadcastState);
        }
    }

    /**
     * 处理=管控命令
     * @param controlType
     * @param rulesState
     * @throws Exception
     */
    private void handleControlCommand(
            Rule.ControlType controlType, BroadcastState<Integer, Rule> rulesState) throws Exception {
        if (controlType == Rule.ControlType.DELETE_RULES_ALL) {
            Iterator<Map.Entry<Integer, Rule>> entriesIterator = rulesState.iterator();
            while (entriesIterator.hasNext()) {
                Map.Entry<Integer, Rule> ruleEntry = entriesIterator.next();
                rulesState.remove(ruleEntry.getKey());
                logger.trace("Removed {}", ruleEntry.getValue());
            }
        }
    }
}
